#include "schedule.h"
#include "log.h"
#include "macro.h"
#include "hook.h"

namespace johnsonli  
{
	static johnsonli::Logger::ptr g_logger = LOG_NAME("system");
	
	//当前线程的调度器
	static thread_local Scheduler* t_scheduler = nullptr;
	//当前调度器的调度协程
	static thread_local Fiber* t_scheduler_fiber = nullptr;

	Scheduler::Scheduler(size_t threads, bool use_caller, const std::string& name)
		:m_name(name)
	{
		DO_ASSERT(threads > 0);
		
		//使用当前线程来运行Scheduler::run，创建Scheduler的调度协程，执行Schedule::run
		if (use_caller)
		{
			johnsonli::Fiber::GetThis();//在当前线程中创建主协程
			--threads; //使用了当前线程执行Scheduler::run，线程池就可以少创建一个
			
			DO_ASSERT(GetThis() == nullptr);//线程创建Scheduler时，t_scheduler还是nullptr，即每个线程只允许有一个调度器
			SetThis(this);
			
			//主线程-->主协程-->调度协程
			//Scheduler的调度协程，用来执行Scheduler的run(调度函数)，stop的时候处理
			m_schedulFiber.reset(new johnsonli::Fiber(std::bind(&Scheduler::run, this), 0, true));
			johnsonli::Thread::SetName(m_name);
			
			t_scheduler_fiber = m_schedulFiber.get();//Scheduler的任务协程
			m_rootThread = johnsonli::getThreadId();//当前线程(主线程)id
			m_threadIds.push_back(m_rootThread);//将当前线程的id放到m_threadIds
		}
		else
		{
			m_rootThread = -1;//当前线程不参与Scheduler::run的调用
		}

		m_threadCount = threads;//线程池中线程数量

	}

	Scheduler::~Scheduler() 
	{
		DO_ASSERT(m_stopping);
		//只在主线程才析构
		if (GetThis() == this) //主线程中创建sc，子线程中又有自己的Schedule，所以必须看这个sc和当前sc是否一样，
		{
			t_scheduler = nullptr;
		}
	}

	std::ostream& Scheduler::dump(std::ostream& os) {
		os << "[Scheduler name=" << m_name
		<< " size=" << m_threadCount
		<< " active_count=" << m_activeThreadCount
		<< " idle_count=" << m_idleThreadCount
		<< " stopping=" << m_stopping
		<< " ]" << std::endl << "    ";
		for(size_t i = 0; i < m_threadIds.size(); ++i) {
			if(i) {
				os << ", ";
			}
			os << m_threadIds[i];
		}
		return os;
	}


	Scheduler* Scheduler::GetThis()
	{
		return t_scheduler;
	}

	void Scheduler::SetThis(Scheduler* s) {
		t_scheduler = s;
	}

	Fiber* Scheduler::GetScheduleFiber()
	{
		return t_scheduler_fiber;
	}
	
	void Scheduler::start()
	{
		
		MutexType::Lock lock(m_mutex);
		if (!m_stopping) { //没有停止，说明在启动，不用启动
			return;
		}
		m_stopping = false;//正在启动

		DO_ASSERT(m_threads.empty());//刚开始。还没有线程
		
		m_threads.resize(m_threadCount);
		for (size_t i = 0; i < m_threadCount; ++i)
		{
			m_threads[i].reset(new Thread(std::bind(&Scheduler::run, this), m_name + "_" + std::to_string(i)));
			m_threadIds.push_back(m_threads[i]->getId());//这里需要注意，当new Thread的时候，我们wait了一下，等线程执行函数初始化完再notify，这样就保证了能拿到此处的线程id;
		}

	}

	void Scheduler::stop()
	{
		m_autoStop = true;

		//调度协程!=null && 调度协程状态=TERM(已经完成) | INIT(还未开始) && 当前线程池为null
		if (m_schedulFiber
			&& (m_schedulFiber->getState() == Fiber::TERM || m_schedulFiber->getState() == Fiber::EXCEPT || m_schedulFiber->getState() == Fiber::INIT)
			&& m_threadCount == 0)
		{
//			LOG_INFO(g_logger) << this << " stopped";
			//m_stopping = true;

			if (stopping())
			{
			//	LOG_INFO(g_logger) << this << " stopped";
				return;
			}
		}

		//use_caller，使用主线程时，只在主线程中stop
		if (m_rootThread != -1) //m_scheduleFiber
		{
			DO_ASSERT(GetThis() == this);
		}
		else
		{
			//不使用主线程，任意线程都可以stop
			DO_ASSERT(GetThis() != this);
		}

		m_stopping = true;

		//唤醒线程池中的线程
		for (size_t i = 0; i < m_threadCount; ++i)
		{
			tickle();
		}

		//唤醒调度协程
		if (m_schedulFiber)
		{
			tickle();
		}

		//调度协程 != null 有m_scheduleFiber 一定是在主线程
		if (m_schedulFiber)
		{
			// if (!stopping())
			// {
			// 	m_schedulFiber->call();
			// }
			m_schedulFiber->call();
		}

		std::vector<Thread::ptr> thrs;
		{
			MutexType::Lock lock(m_mutex);
			thrs.swap(m_threads);
		}

		//让其他线程先结束，留一个线程来回收
		//如果在主线程stop，可能主线程先结束，然后sc就释放，但是子线程还在跑，使用sc，会core dump
		for (auto& i : thrs) {
			i->join();
		}
		
	}

	void Scheduler::run()
	{
		johnsonli::set_hook_enable(true);
		SetThis(this);//给当前线程设置sc
		
		if (johnsonli::getThreadId() != m_rootThread)//当前线程不是主线程
		{
			t_scheduler_fiber = Fiber::GetThis().get();//调度函数 == 当前线程的当前协程
		}

		//LOG_DEBUG(g_logger) << m_name << " run";

		Fiber::ptr idle_fiber(new Fiber(std::bind(&Scheduler::idle, this)));//闲置协程
		Fiber::ptr cb_fiber;//使用回调函数的协程

		FiberAndCallBack fc;
		while (true)
		{
			fc.reset();
			bool tickle_me = false;//是否要通知调度器有任务了
			bool is_active = false;//

			//找到一个当前线程能执行的协程任务
			{ 
				MutexType::Lock lock(m_mutex);
				auto it = m_fibers.begin();
				while (it != m_fibers.end())
				{
					//当前协程任务(FiberAndThread)中设置的线程id != 当前线程id（我们指定了每个协程任务(FiberAndThread)应该在哪里跑）
					if (it->thread != -1 && it->thread != johnsonli::getThreadId())
					{
						++it;
						tickle_me = true;
						continue;
					}

					DO_ASSERT(it->fiber || it->cb);
					//协程任务(FiberAndThread)中的协程目前正在EXEC状态，不执行
					if (it->fiber && it->fiber->getState() == Fiber::EXEC)
					{
						++it;
						continue;
					}

					//找到了一个可以执行的协程任务(FiberAndThread)
					fc = *it;
					m_fibers.erase(it++);//获取当前协程任务(FiberAndThread)后，从协程队列中移除
					++m_activeThreadCount;//增加一个线程执行
					is_active = true;//当前线程存活
					break;
				}

				tickle_me |= (it != m_fibers.end());//如果还没有遍历完，就可以通知调度器有任务
			}
			
			if (tickle_me)
			{
				tickle();//通知调度器有任务
			}

			//当前协程任务(FiberAndThread)中有协程fiber，并且当前协程的状态不是TERM(结束)和EXCEPT(异常)
			if (fc.fiber && (fc.fiber->getState() != Fiber::TERM
				&& fc.fiber->getState() != Fiber::EXCEPT))
			{
				fc.fiber->swapIn();//执行当前协程
				--m_activeThreadCount;//当前协程执行完，说明当前存活线程-1
				
				//当前协程状态为Ready，重新放到协程任务队列
				if (fc.fiber->getState() == Fiber::READY)
				{
					schedule(fc.fiber);
				}
				else if (fc.fiber->getState() != Fiber::TERM
					&& fc.fiber->getState() != Fiber::EXCEPT)
				{
					//当前协程状态不是TERM、EXCEPT，需要改成HOLD
					fc.fiber->setState(Fiber::HOLD);
					//schedule(ft.fiber);
				}

				fc.reset();
			}
			else if (fc.cb)//协程任务有回调函数
			{
				if (cb_fiber)//有
				{
					cb_fiber->reset(fc.cb);
				}
				else//没有
				{
					cb_fiber.reset(new Fiber(fc.cb));
				}
				fc.reset();
				cb_fiber->swapIn();
				--m_activeThreadCount;
				//当前协程状态是Ready
				if (cb_fiber->getState() == Fiber::READY)
				{
					schedule(cb_fiber);
					cb_fiber.reset();
				}
				else if (cb_fiber->getState() == Fiber::EXCEPT
					|| cb_fiber->getState() == Fiber::TERM)
				{
					//状态是TERM(结束) || EXCEPT(异常终止)
					cb_fiber->reset(nullptr);
				}
				else //INIT | HOLD
				{
					cb_fiber->setState(Fiber::HOLD);
					//schedule(cb_fiber);
					cb_fiber.reset();
				}
			}
			else//没有协程，和回调函数，该线程就执行闲置协程的任务
			{
				//m_activeThreadCount已经+1了，需要--
				if (is_active)
				{
					--m_activeThreadCount;
					continue;
				}
				//闲置协程的状态 == TERM(完成)
				if (idle_fiber->getState() == Fiber::TERM) {
					LOG_DEBUG(g_logger) << "idle fiber term";
					break;
				}

				++m_idleThreadCount;
				idle_fiber->swapIn();
				--m_idleThreadCount;

				//限制协程的状态 不是TERM | EXCEPT
				if (idle_fiber->getState() != Fiber::TERM
					&& idle_fiber->getState() != Fiber::EXCEPT) {
					idle_fiber->setState(Fiber::HOLD);
				}
			}
		}

	}

	void Scheduler::tickle() {
		LOG_INFO(g_logger) << "tickle";
	}

	void Scheduler::idle() { 
		LOG_INFO(g_logger) << "idle in"; 
		//如果没结束，就切换成HOLD（暂停）状态，这样就不会退出当前线程
		//这里要使用while，因为下一次swapIn，还会再判断一次
		while(!stopping())
		{
			johnsonli::Fiber::YieldToHoldBySwap();
		}
	}

	bool Scheduler::stopping() 
	{ 
		MutexType::Lock lock(m_mutex);
		//当前协程任务队列 != 空，并且还有线程在运行，就不能停止
		//如果协程任务队列没有任务，就可以停止，没必要执行
		return m_autoStop && m_stopping
			&& m_fibers.empty() && m_activeThreadCount == 0;
	}
}